package com.lrj.datastream.transform.agg;

import com.lrj.datastream.domain.ClickEvent;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author lrj
 * @date 2022/3/28 16:13
 */
public class TestKeyBy {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<ClickEvent> stream = env.fromElements(
                new ClickEvent("Mary", "./home", 1000L),
                new ClickEvent("Blob", "./cart", 2100L),
                new ClickEvent("Mary", "./product?id=100", 3000L),
                new ClickEvent("Mary", "./product?id=1", 3500L),
                new ClickEvent("Mary", "./product?id=23", 200L),
                new ClickEvent("Blob", "./product?id=100", 4000L));
        KeyedStream<ClickEvent, Integer> keyedStream = stream.keyBy(e -> e.getUser().hashCode() % 3);
        // max只更新该字段
        keyedStream.max("timestamp").print("max");
        // maxBy更新所有的字段
        keyedStream.maxBy("timestamp").print("maxBy");
        env.execute(TestKeyBy.class.getName());
    }
}
